package com.wdl.market.app

import com.wdl.market.bean.MarketViewCount
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.sql.Timestamp

class MarketCountResult extends WindowFunction[Long, MarketViewCount, String, TimeWindow ] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[MarketViewCount]): Unit = {
    out.collect(MarketViewCount(
      windowStart = new Timestamp(window.getStart).toString,
      windowEnd = new Timestamp(window.getEnd).toString,
      channel = key,
      behavior = key,
      count = input.head
    ))
  }
}
